package com.clp.inflxudb.core.flux_function;

import com.clp.inflxudb.core.Flux;
import com.clp.inflxudb.core.enums.FluxKey;
import com.clp.inflxudb.core.enums.Func;
import com.clp.inflxudb.core.enums.TsUnit;

/**
 * 指定时间间隔应用聚合函数进行聚合
 */
public class AggregateWindowFluxFunction implements FluxFunction {
    private int interval; // 间隔时间
    private TsUnit tsUnit; // 间隔时间单位
    private Func func; // 聚合函数
    private String keyName; //

    /**
     * @param interval：间隔时间
     * @param tsUnit：间隔时间单位
     * @param func：聚合函数枚举
     * @param keyName：要聚合的键名
     */
    public AggregateWindowFluxFunction(int interval, TsUnit tsUnit, Func func, String keyName) {
        this.interval = interval;
        this.tsUnit = tsUnit;
        this.func = func;
        this.keyName = keyName;
    }

    @Override
    public void doFunction(Flux flux) {
        // 如果时间间隔单位为日及以上，需要转换时区
        if (tsUnit == TsUnit.mo || tsUnit == TsUnit.d || tsUnit == TsUnit.y
                || (tsUnit == TsUnit.h && interval % 24 == 0
                || (tsUnit == TsUnit.m && interval % 1440 == 0) // 1440 = 24 * 60
                || (tsUnit == TsUnit.s && interval % 86400 == 0) // 86400 = 24 * 60 * 60
        )) {
            flux.window(interval, tsUnit, -8, TsUnit.h);
        } else {
            flux.window(interval, tsUnit);
        }
        // 进行聚合
        switch (func) {
            case sum:
                flux.sum(keyName);
                break;
            case last:
                flux.last(keyName);
                break;
            case mean:
                flux.mean(keyName);
                break;
            case first:
                flux.first(keyName);
                break;
            case max:
                flux.max(keyName);
                break;
            case min:
                flux.min(keyName);
                break;
        }
        // 复制time列
        flux.duplicate(FluxKey._stop, FluxKey._time);
        // 合并到无限窗口
        flux.windowInf();
    }
}
